package com.shujia.flink.tf

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo3FlatMap {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.readTextFile("data/words.txt")

    val wordsDS: DataStream[String] = linesDS.flatMap(new FlatMapFunction[String, String] {
      /**
       * flatMap: 传入一行可以返回多行
       *
       * @param value ：一行数据
       * @param out   ： 用于将数据发到下游
       */
      override def flatMap(value: String, out: Collector[String]): Unit = {
        //将一行数据拆分，循环将数据发送到下游
        for (word: String <- value.split(",")) {
          //将数据发送到下游
          out.collect(word)
        }
      }
    })

    wordsDS.print()
    env.execute()
  }

}
